package com.project.mq.mqserver.datacenter;


import com.project.mq.common.MqException;
import com.project.mq.mqserver.core.Message;
import com.project.mq.mqserver.core.Binding;
import com.project.mq.mqserver.core.Exchange;
import com.project.mq.mqserver.core.MSGQueue;

import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 管理所有的内存数据。
 */

public class MemoryDataCenter {
    // key 是 exchangeName,value是Exchange对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key 是 queueName,value是MSGQueue对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 第一个 key 是 exchangeName，第二个 key 是 queueName,value是Binding对象
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // 保存所有消息，key 是 messageId,value是Message对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // key 是 queueName,value是一个Message的链表
    //表示队列和消息之间的关联
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 用来存放待确认的消息
    // key1 是 queueName，key2 是 messageId.
    // 这个结构不需要有对应的硬盘数据。换句话说，如果某个消息消费了，但是没有 ack，这个时候 broker 宕机了，
    // 那么重启 broker 之后就把刚才的消息当做从来没消费过。
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAck = new ConcurrentHashMap<>();

    public void init() {
    }
    //封装 Exchange ⽅法
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功! exchangeName=" + exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功! exchangeName=" + exchangeName);
    }

    //封装 Queue 方法
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 新对列添加成功! queueName=" + queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 对列删除成功! queueName=" + queueName);
    }
    //封装 Binding ⽅法
    public void insertBinding(Binding binding) throws MqException {
        //先使用exchangeName查询一下,对应的哈希表是否存在,不存在就创建一个
        /*ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            bindingMap = new ConcurrentHashMap<>();
            bindingsMap.put(binding.getExchangeName(), bindingMap);
        }*/
        //等价于ConcurrentHashMap<String,Binding> bindingMap =
        // bindingsMap.computeIfAbsent(binding.getExchangeName(),k -> new ConcurrentHashMap<>());
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),k -> new ConcurrentHashMap<>());
        //再根据queueName查一下,如果已经存在,就跳出异常,不催在才能插入
        synchronized (bindingMap) {
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在! exchangeName=" + binding.getExchangeName() + ", queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 新绑定添加成功! exchangeName=" + binding.getExchangeName() + ",queueName=" + binding.getQueueName());
    }
    // 获取绑定，写两个版本：
    // 1. 根据 exchangeName 和 queueName 确定唯一一个 Binding
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);

    }
    // 2. 根据 exchangeName 获取到所有的 Binding
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            //该交换机没有任何绑定,报错
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName() + ", queueName=" + binding.getQueueName());
        }
        synchronized (bindingMap) {
            Binding toDelete = bindingMap.get(binding.getQueueName());
            if (toDelete == null) {
                throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName() + ", queueName=" + binding.getQueueName());
            }
            bindingMap.remove(binding.getQueueName());
        }
        System.out.println("[MemoryDataCenter] 绑定删除成功! exchangeName=" + binding.getExchangeName() + ",queueName=" + binding.getQueueName());
    }
    // 查询指定的消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }
    // 向消息中心添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryCenter] 新消息被添加! messageId=" + message.getMessageId());
    }
    // 从消息中心删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryCenter] 消息被彻底删除! messageId=" + messageId);
    }
    // 发送消息到指定队列中
    public void sendMessage(MSGQueue queue, Message message) {
        //把消息放到对应的队列数据结构中.
        //先根据队列的名字，找到该队列对应的消息链表.
        /*LinkedList<Message> messages = queueMessageMap.get(queue.getName());
        if (messages == null) {
            messages = new LinkedList<>();
            queueMessageMap.put(queue.getName(), messages);
        }*/
        List<Message> messageList = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        synchronized (messageList) {
            messageList.add(message);
        }
        // 在这里把该消息也往消息中心中插入一下。
        // 假设如果 message 已经在消息中心存在，重复插入也没关系，
        // 主要就是相同 messageId，对应的 message 的内容一定是一样的。
        // （服务器代码不会对 Message 内容做修改,对basicPropertices和body不会做修改）
        addMessage(message);
        System.out.println("[MemoryCenter] 消息被投递到队列中! messageId=" + message.getMessageId() + ", queueName=" + queue.getName());
    }
    // 从指定队列中取消息
    public Message pollMessage(String queueName) throws MqException {
        //根据队列名,查找一下,对应的队列的消息链表
        List<Message> messageList = queueMessageMap.get(queueName);
        //如果没有找到,说明队列中没有任何消息
        if (messageList == null) {
            throw new MqException("[MemoryDataCenter] 队列不存在! queueName=" + queueName);
        }
        synchronized (messageList) {
            if (messageList.size() == 0) {
                return null;
            }
            // 出队列头元素
            Message currentMessage = messageList.remove(0);
            System.out.println("[MemoryCenter] 消息从队列中取出! messageId=" + currentMessage.getMessageId() + ", queueName=" + queueName);
            return currentMessage;
        }
    }
    // 获取指定队列中的消息数量
    public int getMessageCount(String queueName) throws MqException {
        List<Message> messageList = queueMessageMap.get(queueName);
        if (messageList == null) {
            // 如果队列不存在，则直接返回长度 0，说明该 queueName 下还没有消息。
            return 0;
        }
        synchronized (messageList) {
            return messageList.size();
        }
    }
    //针对未确认的消息的处理
    // 未被确认的消息，先临时存放一下
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAck.computeIfAbsent(queueName, k -> new ConcurrentHashMap<>());
        synchronized (messageHashMap) {
            messageHashMap.put(message.getMessageId(), message);
        }
        System.out.println("[MemoryCenter] 消息进入待确认队列! messageId=" + message.getMessageId() + ", queueName=" + queueName);
    }
    // 消息被确认之后，就可以真正删除了。
    public void removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAck.get(queueName);
        if (messageHashMap == null) {
            return;
        }
        synchronized (messageHashMap) {
            messageHashMap.remove(messageId);
        }
        System.out.println("[MemoryCenter] 消息从待确认队列删除! messageId=" + messageId + ", queueName=" + queueName);
    }
    //获取指定未确认的消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAck.get(queueName);
        if (messageHashMap == null) {
            return null;
        }
        synchronized (messageHashMap) {
            return messageHashMap.get(messageId);
        }
    }
    //实现重启后恢复内存
    // 这个方法就是从硬盘上读取数据，把硬盘中之前持久化存储的各个维度的数据都恢复到内存中。
    public void recovery(DiskDataCenter diskDataCenter) throws MqException, IOException, ClassNotFoundException {
        //0.清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 1. 恢复交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }

        // 2. 恢复队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }

        // 3. 恢复绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }

        // 4. 恢复消息数据
        // 只需要恢复 queueMessageMap 和 messageMap
        // 注意!! 针对 "未确认的消息" 这部分内存中的数据，不需要从硬盘恢复。之前考虑硬盘存储的时候，也没设定这一块。
        // 一旦在等待 ack 的过程中，服务器重启了，此时这些 "未被确认的消息"，就恢复成 "未被取走的消息"。
        // 这个消息在硬盘上存储的时候，就是当做 “未被取走”。
        //遍历所有的队列,根据每个队列名,读取到所有的消息
        for (MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }
    }

}
